package com.http2;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Http2Client {
    private static SocketChannel sharedChannel;
    private static AtomicInteger nextStreamId = new AtomicInteger(1);
    private static Lock channelLock = new ReentrantLock(); // 防止多线程写冲突
    private static ConcurrentHashMap<Integer, BlockingQueue<Http2Frame>> responseQueues = new ConcurrentHashMap<>();
    
    static {
        try {
            sharedChannel = SocketChannel.open();
            sharedChannel.connect(new InetSocketAddress("localhost", 8080));
            sharedChannel.configureBlocking(false); // 非阻塞模式
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void sendRequest(String payload) throws IOException {
        int streamId = nextStreamId.getAndAdd(2); // HTTP/2客户端Stream ID为奇数
        Http2Frame frame = new Http2Frame();
        frame.streamId = streamId;
        frame.type = 0x1; // HEADERS帧
        frame.payload = payload.getBytes();
        frame.length = frame.payload.length;
        
        // 加锁确保帧完整发送
        channelLock.lock();
        try {
            sharedChannel.write(frame.serialize());
            responseQueues.put(streamId, new ArrayBlockingQueue<>(100));
            System.out.println("发送数据：" + new String(frame.payload) + " Stream " + streamId);
        } finally {
            channelLock.unlock();
        }
        
        // 阻塞队列等待响应
		try {
			Http2Frame response = responseQueues.get(streamId).take();
			System.out.println("Stream " + streamId + " Response: " + new String(response.payload));
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
    }
    
    // 单独启动一个线程处理所有响应
    static {
        new Thread(() -> {
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            while (true) {
                try {
                    if (sharedChannel.read(buffer) > 0) {
                        buffer.flip();
                        Http2Frame frame = Http2Frame.deserialize(buffer);
                        BlockingQueue<Http2Frame> queue = responseQueues.get(frame.streamId);
                        if (queue != null) queue.put(frame); // 按Stream ID分发
                        buffer.clear();
                    }
                } catch (Exception e) {
                	e.printStackTrace();
                }
            }
        }).start();
    }
    
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 5; i++) {
            executor.submit(() -> {
                try {
                    new Http2Client().sendRequest("Request from " + Thread.currentThread().getName());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        //executor.shutdown();
    }
}